package cn.flant.exec;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import cn.flant.data.ArrayMinHeap;
import cn.flant.data.Heap;

public abstract class DelayProcessor<T> {
	public abstract void process(T obj);

	private Processor processor = new Processor();
	private Heap<DelayItem> delayQueue = new ArrayMinHeap<DelayItem>();
	private Lock lock;
	private Condition waiter;
	private boolean isWorking;

	public DelayProcessor(Lock lock) {
		if (lock == null) {
			lock = new ReentrantLock();
		}
		this.lock = lock;
		waiter = lock.newCondition();
	}

	public DelayProcessor() {
		this(new ReentrantLock());
	}

	public void startup() {
		try {
			lock.lock();
			if (isWorking || delayQueue.size() == 0) {
				return;
			}
			isWorking = true;
			ExecutorPool.execute(processor);
		} finally {
			lock.unlock();
		}
	}

	public void shutdown() {
		try {
			lock.lock();
			isWorking = false;
			waiter.signalAll();
		} finally {
			lock.unlock();
		}
	}

	public boolean isWorking() {
		return isWorking;
	}

	public void destroy() {
		try {
			lock.lock();
			isWorking = false;
			delayQueue.clear();
			waiter.signalAll();
		} finally {
			lock.unlock();
		}
	}

	public void submit(T obj, long delay, TimeUnit unit) {
		if (obj == null) {
			return;
		}
		try {
			lock.lock();
			delayQueue.add(new DelayItem(obj, delay, unit));
			waiter.signalAll();
			if (!isWorking) {
				startup();
			}
		} finally {
			lock.unlock();
		}
	}

	public boolean remove(T obj) {
		if (obj == null) {
			return true;
		}
		try {
			lock.lock();
			boolean result = delayQueue.remove(new DelayItem(obj, 0, TimeUnit.MILLISECONDS));
			if (result) {
				waiter.signalAll();
			}
			return result;
		} finally {
			lock.unlock();
		}
	}

	public boolean contains(T obj) {
		if (obj == null) {
			return false;
		}
		try {
			lock.lock();
			return delayQueue.contains(new DelayItem(obj, 0, TimeUnit.MILLISECONDS));
		} finally {
			lock.unlock();
		}
	}

	@SuppressWarnings("unchecked")
	public List<T> getQueueData() {
		DelayItem[] items = null;
		try {
			lock.lock();
			items = delayQueue.toArray(((DelayItem[]) Array.newInstance(DelayItem.class, delayQueue.size())));
		} finally {
			lock.unlock();
		}
		if (items == null) {
			return new ArrayList<T>(0);
		} else {
			List<T> dataList = new ArrayList<T>(items.length);
			for (DelayItem item : items) {
				dataList.add(item.getObject());
			}
			return dataList;
		}

	}

	public Lock getLock() {
		return lock;
	}

	public int getQueueSize() {
		return delayQueue.size();
	}

	public void clearQueue() {
		try {
			lock.lock();
			delayQueue.clear();
			waiter.signalAll();
		} finally {
			lock.unlock();
		}
	}

	private class Processor implements Runnable {
		public void run() {
			while (isWorking) {
				T object = null;
				try {
					lock.lock();
					while (isWorking) {
						DelayItem item = delayQueue.peek();
						if (item == null) {
							isWorking = false;
							return;
						}
						long delay = item.getDelay(TimeUnit.NANOSECONDS);
						if (delay > 0) {
							try {
								waiter.awaitNanos(delay);
							} catch (InterruptedException e) {
							}
							continue;
						} else {
							delayQueue.poll();
							object = item.getObject();
							break;
						}
					}
				} finally {
					lock.unlock();
				}
				if (object == null) {
					continue;
				}
				try {
					process(object);
				} catch (Throwable e) {
				}
			}
		}
	}

	private class DelayItem implements Comparable<DelayItem> {
		private long execTime;
		private T object;

		public DelayItem(T obj, long delay, TimeUnit unit) {
			this.object = obj;
			this.execTime = System.nanoTime() + unit.toNanos(delay);
		}

		public T getObject() {
			return object;
		}

		public int compareTo(DelayItem o) {
			long d = execTime - o.execTime;
			if (d > 0) {
				return 1;
			} else if (d == 0) {
				return 0;
			} else {
				return -1;
			}
		}

		public long getDelay(TimeUnit unit) {
			long delay = execTime - System.nanoTime();
			return unit.convert(delay, TimeUnit.NANOSECONDS);
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result + getOuterType().hashCode();
			result = prime * result + ((object == null) ? 0 : object.hashCode());
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			@SuppressWarnings("unchecked")
			DelayItem other = (DelayItem) obj;
			if (!getOuterType().equals(other.getOuterType()))
				return false;
			if (object == null) {
				if (other.object != null)
					return false;
			} else if (!object.equals(other.object))
				return false;
			return true;
		}

		@SuppressWarnings("rawtypes")
		private DelayProcessor getOuterType() {
			return DelayProcessor.this;
		}
	}
}
